package cn.baiyang.rocketmq.spring.support;

import cn.baiyang.rocketmq.spring.annotation.ConsumerType;
import cn.baiyang.rocketmq.spring.annotation.RocketMQMessageListener;
import cn.baiyang.rocketmq.spring.annotation.SelectorType;
import cn.baiyang.rocketmq.spring.core.RocketMQListener;

import java.time.Duration;
import java.util.Collections;
import java.util.Objects;

import org.apache.rocketmq.client.apis.ClientConfiguration;
import org.apache.rocketmq.client.apis.ClientException;
import org.apache.rocketmq.client.apis.ClientServiceProvider;
import org.apache.rocketmq.client.apis.SessionCredentialsProvider;
import org.apache.rocketmq.client.apis.StaticSessionCredentialsProvider;
import org.apache.rocketmq.client.apis.consumer.FilterExpression;
import org.apache.rocketmq.client.apis.consumer.FilterExpressionType;
import org.apache.rocketmq.client.apis.consumer.PushConsumer;
import org.apache.rocketmq.client.apis.consumer.PushConsumerBuilder;
import org.apache.rocketmq.client.java.impl.ClientImpl;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.SmartLifecycle;
import org.springframework.util.Assert;

/**
 *
 * @author hongzhu
 * @version V1.0
 * @since 2023-07-24 18:58
 */
public class DefaultRocketMQListenerContainer implements InitializingBean,
    RocketMQListenerContainer, SmartLifecycle, ApplicationContextAware {
    private final static Logger LOGGER = LoggerFactory.getLogger(DefaultRocketMQListenerContainer.class);

    private ApplicationContext applicationContext;

    /**
     * name of this container
     */
    private String name;

    /*---------------------------------RocketMQ Listener---------------------------------*/

    private String endpoints;

    private String accessKey;

    private String secretKey;

    private String enableSsl;

    private String requestTimeout;

    private String consumerGroup;

    private String topic;

    private SelectorType selectorType;

    private String selectorExpression;

    private int maxCacheMessageCount;

    private int maxCacheMessageSizeInBytes;

    private int consumptionThreadCount;

    private ConsumerType consumerType;

    private String instanceName;

    /*---------------------------------RocketMQ Runtime---------------------------------*/

    private RocketMQListener rocketMQListener;

    private RocketMQMessageListener rocketMQMessageListener;

    private PushConsumer pushConsumer;

    private boolean running;

    /*=================================Method=================================*/

    public String getName() {
        return name;
    }

    public void setName(String name) {
        this.name = name;
    }

    /*---------------------------------RocketMQ Listener---------------------------------*/

    public String getEndpoints() {
        return endpoints;
    }

    public void setEndpoints(String endpoints) {
        this.endpoints = endpoints;
    }

    public String getAccessKey() {
        return accessKey;
    }

    public void setAccessKey(String accessKey) {
        this.accessKey = accessKey;
    }

    public String getSecretKey() {
        return secretKey;
    }

    public void setSecretKey(String secretKey) {
        this.secretKey = secretKey;
    }

    public String getEnableSsl() {
        return enableSsl;
    }

    public void setEnableSsl(String enableSsl) {
        this.enableSsl = enableSsl;
    }
    public String getRequestTimeout() {
        return requestTimeout;
    }

    public void setRequestTimeout(String requestTimeout) {
        this.requestTimeout = requestTimeout;
    }

    public String getConsumerGroup() {
        return consumerGroup;
    }

    public void setConsumerGroup(String consumerGroup) {
        this.consumerGroup = consumerGroup;
    }

    public String getTopic() {
        return topic;
    }

    public void setTopic(String topic) {
        this.topic = topic;
    }

    public SelectorType getSelectorType() {
        return selectorType;
    }

    public void setSelectorType(SelectorType selectorType) {
        this.selectorType = selectorType;
    }

    public String getSelectorExpression() {
        return selectorExpression;
    }

    public void setSelectorExpression(String selectorExpression) {
        this.selectorExpression = selectorExpression;
    }

    public int getMaxCacheMessageCount() {
        return maxCacheMessageCount;
    }

    public void setMaxCacheMessageCount(int maxCacheMessageCount) {
        this.maxCacheMessageCount = maxCacheMessageCount;
    }

    public int getMaxCacheMessageSizeInBytes() {
        return maxCacheMessageSizeInBytes;
    }

    public void setMaxCacheMessageSizeInBytes(int maxCacheMessageSizeInBytes) {
        this.maxCacheMessageSizeInBytes = maxCacheMessageSizeInBytes;
    }

    public int getConsumptionThreadCount() {
        return consumptionThreadCount;
    }

    public void setConsumptionThreadCount(int consumptionThreadCount) {
        this.consumptionThreadCount = consumptionThreadCount;
    }

    public ConsumerType getConsumerType() {
        return consumerType;
    }

    public void setConsumerType(ConsumerType consumerType) {
        this.consumerType = consumerType;
    }

    public String getInstanceName() {
        return instanceName;
    }

    public void setInstanceName(String instanceName) {
        this.instanceName = instanceName;
    }
    /*---------------------------------RocketMQ Runtime---------------------------------*/

    public RocketMQListener getRocketMQListener() {
        return rocketMQListener;
    }

    public void setRocketMQListener(RocketMQListener rocketMQListener) {
        this.rocketMQListener = rocketMQListener;
    }

    public RocketMQMessageListener getRocketMQMessageListener() {
        return rocketMQMessageListener;
    }

    public PushConsumer getPushConsumer() {
        return pushConsumer;
    }

    public void setPushConsumer(PushConsumer pushConsumer) {
        this.pushConsumer = pushConsumer;
    }

    public void setRocketMQMessageListener(RocketMQMessageListener anno) {
        this.rocketMQMessageListener = anno;

        this.endpoints = anno.endpoints();
        this.accessKey = anno.accessKey();
        this.secretKey = anno.secretKey();
        this.enableSsl = anno.enableSsl();
        this.requestTimeout = anno.requestTimeout();
        this.topic = anno.topic();
        this.consumerGroup = anno.consumerGroup();
        this.selectorType = anno.selectorType();
        this.selectorExpression = anno.selectorExpression();
        this.consumptionThreadCount = anno.consumptionThreadCount();
        this.maxCacheMessageCount = anno.maxCacheMessageCount();
        this.maxCacheMessageSizeInBytes = anno.maxCacheMessageSizeInBytes();
        this.consumerType = anno.consumerType();
        this.instanceName = anno.instanceName();
    }

    @Override
    public void destroy() {
        this.setRunning(false);

        switch (consumerType) {
        case PUSH:
            if (Objects.nonNull(pushConsumer)) {
                ((ClientImpl) pushConsumer).stopAsync().awaitTerminated();
            }
            break;
        case SIMPLE:
            // TODO
            break;
        case PULL:
            // TODO
            break;
        default:
        }

        LOGGER.info("container destroyed, {}", this.toString());
    }

    @Override
    public boolean isAutoStartup() {
        return true;
    }

    @Override
    public void stop(Runnable callback) {
        stop();
        callback.run();
    }

    @Override
    public void start() {
        if (this.isRunning()) {
            throw new IllegalStateException("container already running. " + this.toString());
        }

        switch (consumerType) {
        case PUSH:
            ((ClientImpl) pushConsumer).startAsync().awaitRunning();
            break;
        case SIMPLE:
            // TODO
            break;
        case PULL:
            // TODO
            break;
        default:
        }
        this.setRunning(true);

        LOGGER.info("started container, {}", this.toString());
    }

    @Override
    public void stop() {
        if (!this.isRunning()) {
            return;
        }

        switch (consumerType) {
        case PUSH:
            if (Objects.nonNull(pushConsumer)) {
                ((ClientImpl) pushConsumer).stopAsync().awaitTerminated();
            }
            break;
        case SIMPLE:
            // TODO
            break;
        case PULL:
            // TODO
            break;
        default:
        }

        this.setRunning(false);
        LOGGER.info("stopped container, {}", this.toString());
    }

    @Override
    public boolean isRunning() {
        return running;
    }

    private void setRunning(boolean running) {
        this.running = running;
    }

    @Override
    public int getPhase() {
        // Returning Integer.MAX_VALUE only suggests that
        // we will be the first bean to shutdown and last bean to start
        return Integer.MAX_VALUE;
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        initRocketMQPushConsumer();
    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
    }

    @Override
    public String toString() {
        return "DefaultRocketMQListenerContainer{" +
            "name='" + name + '\'' +
            ", endpoints='" + endpoints + '\'' +
            ", consumerGroup='" + consumerGroup + '\'' +
            ", topic='" + topic + '\'' +
            ", selectorType=" + selectorType +
            ", selectorExpression='" + selectorExpression + '\'' +
            ", instanceName='" + instanceName + '\'' +
            '}';
    }

    private void initRocketMQPushConsumer() throws ClientException {
        if (rocketMQListener == null) {
            throw new IllegalArgumentException("Property 'rocketMQListener' is required");
        }
        Assert.hasText(endpoints, "Property 'endpoints' is required");
        Assert.hasText(consumerGroup, "Property 'consumerGroup' is required");
        Assert.hasText(topic, "Property 'topic' is required");

        FilterExpression filterExpression;
        switch (selectorType) {
        case TAG:
            filterExpression = new FilterExpression(selectorExpression, FilterExpressionType.TAG);
            break;
        case SQL92:
            filterExpression = new FilterExpression(selectorExpression, FilterExpressionType.SQL92);
            break;
        default:
            throw new IllegalArgumentException("Property 'selectorType' was wrong, expect TAG or SQL92");
        }

        SessionCredentialsProvider sessionCredentialsProvider = new StaticSessionCredentialsProvider(accessKey, secretKey);
        ClientConfiguration clientConfiguration = ClientConfiguration.newBuilder()
            .setEndpoints(endpoints)
            .setRequestTimeout(Duration.ofSeconds(Integer.parseInt(requestTimeout)))
            .enableSsl(Boolean.parseBoolean(enableSsl))
            .setCredentialProvider(sessionCredentialsProvider)
            .build();
        ClientServiceProvider provider = ClientServiceProvider.loadService();

        switch (consumerType) {
        case PUSH:
            PushConsumerBuilder pushConsumerBuilder = provider.newPushConsumerBuilder()
                .setClientConfiguration(clientConfiguration)
                .setConsumerGroup(consumerGroup)
                .setSubscriptionExpressions(Collections.singletonMap(topic, filterExpression))
                .setConsumptionThreadCount(consumptionThreadCount)
                .setMaxCacheMessageCount(maxCacheMessageCount)
                .setMaxCacheMessageSizeInBytes(maxCacheMessageSizeInBytes)
                .setMessageListener(rocketMQListener);
            pushConsumer = pushConsumerBuilder.build();
            break;
        case SIMPLE:
            throw new IllegalArgumentException("SimpleConsumer not supported now");
        case PULL:
            throw new IllegalArgumentException("PullConsumer not supported now");
        default:
            throw new IllegalArgumentException("Unknown Consumer Type. Selection is Push, Simple, Pull");
        }
        this.setRunning(true);
    }

}
